package cas.ihep.pwa.fcn;

import alluxio.AlluxioURI;
import cas.ihep.alluxio.AlluxioBlock;
import cas.ihep.alluxio.AlluxioFile;
import cas.ihep.alluxio.AlluxioSystem;
import cas.ihep.pwa.entity.ComputeResultMc;
import cas.ihep.pwa.evtgen.EvtPDL;
import cas.ihep.spark.JobParser;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import scala.Tuple2;
import scala.Tuple3;

import java.util.ArrayList;
import java.util.Iterator;

public class McFcn extends BaseFcn implements PairFlatMapFunction<String,Long,ComputeResultMc> {

    private static final long serialVersionUID=0xcafebabe9264342L;
    public McFcn(){super();}

//    public McFcn(Broadcast<EvtPDL> pdl_,Broadcast<JobParser> p,Broadcast<double[]> v,Broadcast<String> decay){
//        super(pdl_,p,v,decay);
//    }

    public McFcn(EvtPDL pdl,JobParser jp,double []v,String d){
        super(pdl,jp,v,d);
    }

    public Tuple3<Double, Double, double[]> call0(String tuple) throws Exception {

        double ret[]=new double[32];


        return new Tuple3<>(ret[30],ret[31],ret);
    }

    public Iterator<Tuple2<Long, ComputeResultMc>> call(String args)throws Exception{
        mmapDecay(bdecay);
        AlluxioSystem asys=AlluxioSystem.get();
        AlluxioURI apath=new AlluxioURI(args);
        AlluxioFile afile=asys.open(apath);
        System.out.println("Processing MC file: "+args);
        AlluxioBlock ablk=afile.getBlock(0);
        String blkpath=ablk.getBlockPath();
        ArrayList<Tuple2<Long,ComputeResultMc>> retlst=new ArrayList<>();
        blkpath=thcalamps(blkpath,vpars,retlst,pdl,parser,parser);
        ablk.close();
        afile.close();
        asys.getFileSystem().delete(apath);
        afile=asys.create(apath);
        ablk=afile.nextBlock();
        link(blkpath,ablk.getBlockPath());
        ablk.close();
        afile.close();
        return retlst.iterator();
    }
}
